package com.blackList

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  *
  * @author ymy.hadoop
  *
  */
object BlackListDemo {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("BlackListDemo").setMaster("local[2]")
    val blackList = List("zs","ls")

    val ssc = new StreamingContext(sparkConf,Seconds(10))
    // 构建黑名单("xxx",true)
    val blackRDD = ssc.sparkContext.parallelize(blackList).map(x=>(x,true))
    val lines = ssc.socketTextStream("localhost",9000)

   // ("xxx","20190806","xxx") leftOurerJoin ("xxx",true)
    val clicklog = lines.map(x=>(x.split(",")(1),x)).transform(rdd=>{
     // join后变为("xxx",(("20190806","xxx"),true或者是None))
      rdd.leftOuterJoin(blackRDD)
        .filter(x=>x._2._2.getOrElse(false)!=true)
        .map(x=>x._2._1)
    }).foreachRDD(_.foreach(println))

    ssc.start()
    ssc.awaitTermination()
  }
}
